package yu.code.yurpc.registry;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.ConcurrentHashSet;
import cn.hutool.cron.CronUtil;
import cn.hutool.cron.task.Task;
import cn.hutool.json.JSONUtil;
import io.etcd.jetcd.*;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.PutOption;
import io.etcd.jetcd.watch.WatchEvent;
import io.grpc.internal.JsonUtil;
import yu.code.yurpc.config.RegistryConfig;
import yu.code.yurpc.listener.EtcdRegistryListener;
import yu.code.yurpc.listener.RegistryListener;
import yu.code.yurpc.model.ServiceMetaInfo;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

/**
 * Etcd 注册中心实现
 */
public class EtcdRegistry implements Registry {
    /**
     * 本机注册的节点 key 集合（用于维护续期）
     */
    private final Set<String> localRegisterNodeKeySet = new HashSet<>();

    private Client client;
    private KV kvClient;
    /**
     * etcd 根节点
     */
    public static final String ETCD_ROOT_PATH = "/rpc/";
    /**
     * 正在监听的 key 集合
     */
    private final Set<String> watchingKeySet = new ConcurrentHashSet<>();
    /**
     * 本地服务缓存
     */
    RegistryServiceCache registryServiceCache = new RegistryServiceCache();

    @Override
    public void init(RegistryConfig registryConfig) {
        client = Client.builder().endpoints(registryConfig.getAddress()).connectTimeout(Duration.ofMillis(registryConfig.getTimeout())).build();
        kvClient = client.getKVClient();
        heartbeat();
    }

    @Override
    public void register(ServiceMetaInfo serviceMetaInfo) throws Exception {
        //创建Lease和KV客户端
        Lease lease = client.getLeaseClient();
        //创建一个30秒的租约
        long leaseId = lease.grant(600).get().getID();
        //设置要存储的键值对
        String registerKey = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        ByteSequence key = ByteSequence.from(registerKey, StandardCharsets.UTF_8);
        ByteSequence value = ByteSequence.from(JSONUtil.toJsonStr(serviceMetaInfo), StandardCharsets.UTF_8);
        //将键值对绑定到租约上
        PutOption putOption = PutOption.builder().withLeaseId(leaseId).build();
        //添加到本地已注册节点集合
        localRegisterNodeKeySet.add(registerKey);
        kvClient.put(key, value, putOption).get();
    }

    @Override
    public void unRegister(ServiceMetaInfo serviceMetaInfo) throws Exception {
        String registerKey = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        kvClient.delete(ByteSequence.from(registerKey, StandardCharsets.UTF_8)).get();
        //移除本地已注册节点集合
        localRegisterNodeKeySet.remove(registerKey);
    }

    /**
     * 获取注册中心服务列表
     *
     * @param serviceKey
     * @return
     */
    @Override
    public List<ServiceMetaInfo> getServiceMetaInfoList(String serviceKey) {
        //先查缓存
        if (registryServiceCache.readCache() != null) {
            return registryServiceCache.readCache();
        }
        //前缀搜索
        String searchPrefix = ETCD_ROOT_PATH + serviceKey + "/";
        try {
            //前缀查询
            GetOption getOption = GetOption.builder().isPrefix(true).build();

            List<KeyValue> kvs = null;
            try {
                kvs  = kvClient.get(ByteSequence.from(searchPrefix, StandardCharsets.UTF_8), getOption).get().getKvs();

            }catch (Exception e){
                e.printStackTrace();
                System.out.println("");
                throw e;
            }
            //解析服务信息
            List<ServiceMetaInfo> serviceMetaInfos = kvs.stream().map(kv -> {
                String value = kv.getValue().toString(StandardCharsets.UTF_8);
                return JSONUtil.toBean(value, ServiceMetaInfo.class);
            }).collect(Collectors.toList());
            //写入缓存
            registryServiceCache.writeCache(serviceMetaInfos);
            return serviceMetaInfos;
        } catch (Exception e) {
            throw new RuntimeException("获取服务列表失败", e);
        }
    }

    @Override
    public void destroy() {
        System.out.println("当前节点下线");
        //遍历本节点所有的key
        for (String registerNodeKey : localRegisterNodeKeySet) {
            try {
                //删除节点
                kvClient.delete(ByteSequence.from(registerNodeKey, StandardCharsets.UTF_8)).get();
            } catch (Exception e) {
                throw new RuntimeException(registerNodeKey + "节点下线失败", e);
            }
        }
        // 释放资源
        if (kvClient != null) {
            kvClient.close();
        }
        if (client != null) {
            client.close();
        }
    }

    @Override
    public void heartbeat() {
        //10秒续签一次
        CronUtil.schedule("0/10 * * * * ?", new Task() {
            @Override
            public void execute() {
                //遍历本地以注册的节点
                for (String registerNodeKey : localRegisterNodeKeySet) {
                    //获取节点
                    try {
                        List<KeyValue> kvs = kvClient.get(ByteSequence.from(registerNodeKey, StandardCharsets.UTF_8)).get().getKvs();
                        if (CollUtil.isEmpty(kvs)) {
                            //当前节点已经过期（需要重新启动注册）
                            continue;
                        }
                        //节点未过期，重新注册（续期）
                        KeyValue keyValue = kvs.get(0);
                        String value = keyValue.getValue().toString(StandardCharsets.UTF_8);
                        ServiceMetaInfo serviceMetaInfo = JSONUtil.toBean(value, ServiceMetaInfo.class);
                        register(serviceMetaInfo);
                    } catch (Exception e) {
                        throw new RuntimeException(registerNodeKey + "续签失败", e);
                    }
                }
            }
        });
        //支持秒级定时任务
        CronUtil.setMatchSecond(true);
        CronUtil.start();
    }

    /**
     * 监听（消费者）
     *
     * @param serviceKey
     */
    @Override
    public void watch(String serviceKey) {
        Watch watchClient = client.getWatchClient();
        //之前未监听该key，则监听
        boolean newWatch = watchingKeySet.add(serviceKey);
        //绑定监听器
        RegistryListener registryListener = new EtcdRegistryListener(registryServiceCache);
        if (newWatch) {
            watchClient.watch(ByteSequence.from(serviceKey, StandardCharsets.UTF_8), watchResponse -> {
                for (WatchEvent event : watchResponse.getEvents()) {
                    registryListener.operateCache(event);
                }
            });
        }
    }
}
